package cn.doitedu.dmp

import java.util

import org.apache.commons.lang3.StringUtils
import org.apache.log4j.{Level, Logger}
import org.apache.spark.graphx.{Edge, Graph}
import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{Dataset, SparkSession}

object DmpLogIdMapping {
  def main(args: Array[String]): Unit = {


    Logger.getLogger("org").setLevel(Level.WARN)

    val spark = SparkSession.builder()
      .appName("DMP日志用户标识IDMAPPING映射字典生成")
      .master("local")
      .getOrCreate()

    import spark.implicits._

    // 加载数据
    val ds: Dataset[String] = spark.read.textFile("user_portrait/data/dmp_log")

    // 提取用户标识
    val flagsArrRdd: RDD[List[String]] = ds.rdd.map(line=>{
      val strings: Array[String] = line.split(",")
      val flags = strings.slice(46, 51).filter(StringUtils.isNotBlank(_))
      flags.toList
    }).filter(_.size>0)

    // 将字符串标识，转化成图计算中的点（Vertex）
    val verticesListRdd: RDD[List[(Long, String)]] = flagsArrRdd.map(lst=>{
      for (elem <- lst) yield (elem.hashCode.toLong,elem)
    })

    // 将 "点list" 集合，打散成：点集合
    val verticesRdd = verticesListRdd.flatMap(lst=>lst).distinct()

    // 生成“边”集合
    val edgesRdd: RDD[Edge[String]] = verticesListRdd.flatMap(vertexList=>{
      val firstVertex = vertexList(0)
      for (i <- 1 until vertexList.size) yield  Edge(firstVertex._1,vertexList(i)._1,"")
    })


    // 将 点集合  +  边集合   构造成一个图对象Graph
    val graph = Graph(verticesRdd, edgesRdd)

    // 在图上调算法：最大连通子图
    val resultGraph = graph.connectedComponents()

    // 从结果图中，取出结果图的“点集合”, 这个点集合有名堂：每一个点都带了一个属性（同一个连通子图点，带的属性相同——是这一组点中的最小id值）
    val resultVertices = resultGraph.vertices

    resultVertices.toDF("flagId","connectedId").write.parquet("user_portrait/data/dmp_log_idmpdict")

    spark.close()

  }

}
